package t20240724;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;


/**
 last_value：
   返回最后一个非null值。

 */
public class D5_Dump {


    public static void main(String[] args) throws InterruptedException {


        Configuration flinkConf = new Configuration();
        flinkConf.setString("rest.port","9092");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        String genSql = "CREATE TABLE ods_tb ( " +
                " stime TIMESTAMP(3)," +
                " name STRING," +
                " id BIGINT," +
                " ts AS localtimestamp," +
                " rowtime AS localtimestamp," +
                " WATERMARK FOR rowtime AS rowtime - interval '30' second" +
                ") WITH ( " +
                " 'connector'='datagen', " +
                " 'rows-per-second'='2'," +
                " 'fields.name.length'='1'," +
                " 'fields.id.kind'='random'" +
                ")";

        String sink1 = "CREATE TABLE sink1 (" +"    " +
                "    stime TIMESTAMP(3), " +
                "    name STRING, " +
                "    id BIGINT, " +
                "    primary key (stime) not ENFORCED" +
                ") WITH (" +
                "    'database-name' = 'default',\n" +
                "    'password' = '',\n" +
                "    'use-local' = 'false',\n" +
                "    'connector' = 'clickhouse',\n" +
                "    'sink.flush-interval' = '10s',\n" +
                "    'sink.batch-size' = '10000',\n" +
                "    'url' = 'clickhouse://127.0.0.1:8123',\n" +
                "    'table-name' = 't1',\n" +
                "    'username' = 'default'" +
                ")";


        String sink2 = "CREATE TABLE sink2 (" +"    " +
                "    stime TIMESTAMP(3), " +
                "    name STRING, " +
                "    id BIGINT " +
//                "    primary key (stime) not ENFORCED" +
                ") WITH (" +
                "     'database-name' = 'default',\n" +
                "    'password' = '',\n" +
                "    'use-local' = 'false',\n" +
                "    'connector' = 'clickhouse',\n" +
                "    'sink.flush-interval' = '10s',\n" +
                "    'sink.batch-size' = '10000',\n" +
                "    'url' = 'clickhouse://127.0.0.1:8123',\n" +
                "    'table-name' = 't2',\n" +
                "    'username' = 'default'" +
                ")";

        String sql1 = "insert into sink1 " +
                " SELECT stime,name,id FROM " +
                " (SELECT *,ROW_NUMBER() OVER(PARTITION BY stime,name ORDER BY rowtime) as rownum FROM ods_tb) " +
                " WHERE rownum = 1";

        String sql2 = "insert into sink2 " +
                " SELECT stime,name,id FROM ods_tb";


        tableEnv.executeSql(genSql);

        tableEnv.executeSql(sink1);
        tableEnv.executeSql(sink2);

        tableEnv.executeSql(sql1);
        tableEnv.executeSql(sql2);

        System.out.println(new Timestamp(System.currentTimeMillis()));



    }
}
